#
# Copyright 2009 Red Hat, Inc. and/or its affiliates.
#
# Licensed to you under the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.  See the files README and
# LICENSE_GPL_v2 which accompany this distribution.
#

'''
    Tasks: tasks object model some sort of VDSM task (storage task).
    A task object may be standalone (unmanaged task), but then it is limited and
    cannot be automatically persisted or run (asyncronous jobs).
    A managed task is managed by a TaskManager. Under the task manager a task may
    persist itself, run asynchronous jobs, run recovery procedures, etc.
    A task is successful if it finished its operations during the
    prepare state (i.e. it did not schedule any async jobs), or all its scheduled
    jobs complete without error.
    The task result is the prepare state result if no async jobs are scheduled,
    or the result of the last job run.

    Jobs: jobs object model an asyncronous job that is run in a context of a
    thread belonging to a thread pool (and managed by the task manager). Currently
    a task may schedule any number of jobs that run sequentially by the same
    worker thread.

    Recovery: recovery objects model a recovery operation required to restore
    the system to a known coherent state. A task may register several recovery
    objects that are kept in a stack (the last recovery registered is run first - lifo).
    Tasks that have "auto" recovery policy run the recovery procedue ("rollback"),
    in any case of task failure/abort, and immediately after the task is
    recovered (loaded) from its persisted store.
'''

import os
import logging
import threading
import glob

import storage_exception as se
import misc
import uuid
import resource
from threadLocal import vars
from weakref import proxy
from config import config

from remoteFileHandler import remoteFileHandlerPool
ros = remoteFileHandlerPool

KEY_SEPERATOR = "="
TASK_EXT = ".task"
JOB_EXT = ".job"
RESOURCE_EXT = ".resource"
RECOVER_EXT = ".recover"
RESULT_EXT = ".result"
BACKUP_EXT = ".backup"
TEMP_EXT = ".temp"
NUM_SEP = "."
FIELD_SEP = ","
RESOURCE_SEP = "!"
TASK_METADATA_VERSION = 1


class State:
    unknown = "unknown"
    init = "init"
    preparing = "preparing"
    blocked = "blocked"
    acquiring = "acquiring"
    queued = "queued"
    running = "running"
    finished = "finished"
    aborting = "aborting"
    waitrecover = "waitrecover"
    recovering = "recovering"
    racquiring = "racquiring"
    raborting = "raborting"
    recovered = "recovered"
    failed = "failed"

    # For backward compatibility, inner states can be translated to previously supported
    # format of state and result
    DEPRECATED_STATE = {
        unknown: "unknown",
        init: "init",
        preparing: "running",
        blocked: "running",
        acquiring: "running",
        queued: "running",
        running: "running",
        finished: "finished",
        aborting: "aborting",
        waitrecover: "cleaning",
        recovering: "cleaning",
        racquiring: "cleaning",
        raborting: "aborting",
        recovered: "finished",
        failed: "finished"
    }

    DEPRECATED_RESULT = {
        unknown: "",
        init: "",
        preparing: "",
        blocked: "",
        acquiring: "",
        queued: "",
        running: "",
        finished: "success",
        aborting: "",
        waitrecover: "",
        recovering: "",
        racquiring: "",
        raborting: "",
        recovered: "cleanSuccess",
        failed: "cleanFailure"
    }

    # valid state transitions: newstate <- [from states]
    _moveto = {
        unknown: [],
        init: [],
        preparing: [init, blocked],
        blocked: [preparing],
        acquiring: [preparing, acquiring],
        queued: [acquiring, running],
        running: [queued],
        finished: [running, preparing],
        aborting: [ preparing, blocked, acquiring, queued, running ],
        waitrecover: [ aborting ],
        racquiring: [aborting, finished, racquiring, waitrecover],
        recovering: [racquiring],
        raborting: [racquiring, recovering, waitrecover],
        recovered: [recovering],
        failed: [recovering, aborting, raborting],
    }
    _done = [finished, recovered, failed]


    def __init__(self, state = None):
        try:
            if not state:
                self.state = self.unknown
            else:
                self.state = getattr(self, state)
        except:
            self.state = self.unknown


    def value(self):
        return self.state


    def isDone(self):
        return self.state in self._done


    def canAbort(self):
        return self.state in self._moveto[self.aborting]

    def canAbortRecovery(self):
        return self.state in self._moveto[self.raborting]


    def canAbortRecover(self):
        return self.state in self._moveto[self.raborting]


    def __str__(self):
        return self.state


    def moveto(self, state, force=False):
        if state not in self._moveto:
            raise ValueError("not a valid target state: %s" % str(state))
        if not force and self.state not in self._moveto[state]:
            raise se.TaskStateTransitionError("from %s to %s" % (self.state, state))
        self.state = state


    def __eq__(self, state):
        return self.state == state


    def __ne__(self, state):
        return self.state != state



class EnumType(object):
    def __init__(self, enum):
        if not getattr(self, enum, None):
            raise ValueError("%s not a valid type for %s" % (enum, repr(self)))
        self.value = enum


    def __str__(self):
        return str(self.value)


    def __eq__(self, x):
        if type(x) == str:
            return self.value == x
        if isinstance(x, self):
            return x.value == self.value
        return False


    def __ne__(self, x):
        return not self.__eq__(x)



class TaskPersistType(EnumType):
    none = "none"
    manual = "manual"
    auto = "auto"



class TaskCleanType(EnumType):
    none = "none"
    manual = "manual"
    auto = "auto"



class TaskRecoveryType(EnumType):
    none = "none"
    manual = "manual"
    auto = "auto"



class BoolStr(EnumType):
    True = "True"
    False = "False"


    def __init__(self, enum):
        if str(enum) == "0":
            enum = self.True
        elif str(enum) == "1":
            enum = self.False
        else:
            enum = enum.capitalize()
        EnumType.__init__(self, enum)


    def __eq__(self, x):
        if type(x) == bool:
            return bool(self.value) == x
        return EnumType.__eq__(self, x)



class ParamList:

    def __init__(self, params="", sep=FIELD_SEP):
        self.sep = sep
        if not params:
            self.params = ""

        if type(params) == list:
            for i in params:
                if type(i) != str:
                    raise ValueError("ParamsList: param item %s not a string (%s)" % (i, type(i)))
                if sep in i:
                    raise ValueError("ParamsList: sep %s in %s" % (sep, i))
            self.params = params
        elif type(params) == str:
            self.params = [ s.strip() for s in params.split(sep) ]
        else:
            raise ValueError("ParamList: params type not supported (%s)" % str(type(params)))


    def getList(self):
        return self.params[:]


    def __str__(self):
        s = ""
        for i in self.params:
            s += str(i) + self.sep
        # remove last sep
        if s:
            s = s[:-1]
        return s



class Job:
    fields = {
        "name": str,
        "runcmd": str,
    }

    def __init__(self, name, cmd, *argslist, **argsdict):
        self.name = name
        self.cmd = cmd       # function pointer to run
        self.argslist = argslist
        self.argsdict = argsdict
        self.runcmd = "%s (args: %s kwargs: %s)" % (repr(cmd), str(argslist), str(argsdict))          # args str
        self.callback = None    # callback to call before running the job
        self.task = None


    def setCallback(self, callback):
        if not callable(callback):
            raise ValueError("Task.Job: callback %s is not callable" % repr(callback))
        self.callback = callback


    def setOwnerTask(self, task):
        self.task = proxy(task)

    def run(self):
        if not self.task:
            raise se.InvalidJob("Job %s: no parent task" % str(self))
        self.task.log.debug("Job.run: running %s callback %s" % (str(self), repr(self.callback)))
        if self.callback:
            self.callback(self)
        return self.cmd(*self.argslist, **self.argsdict)


    def __str__(self):
        return "%s: %s" % (self.name, self.runcmd)



class Recovery:
    '''
    A recovery object is used to register a recovery function to the recovery process.
    The recovery functions are kept in a stack so they are carried out in FILO order.
    The recovery function itself must be a static/class binded method of a visible class.
    The recovery function must accept a recovery object as the first parameter - fn(recovery, ...)
    All other parameters if any must be strings.
    '''
    fields = {
        "name": str,
        "moduleName": str,
        "object": str,
        "function": str,
        "params": ParamList,
    }

    def __init__(self, name, modname, objname, fnname, argslist):
        self.name = name
        self.validateName(modname)
        self.validateName(objname)
        self.validateName(modname)
        self.validateName(fnname)
        self.object = objname
        self.moduleName = modname
        self.function = fnname
        self.params = ParamList(argslist)
        self.callback = None
        self.task = None


    def validateName(self, name):
        vname = name.replace("_", "")
        if not vname.isalnum():
            raise TypeError("Parameter %s must be a plain str" % name)


    def setCallback(self, callback):
        if not callable(callback):
            raise ValueError("Task.Recovery: callback %s is not callable" % repr(callback))
        self.task.log.debug("Recovery.run: running %s callback %s" % (str(self), repr(self.callback)))
        self.callback = callback


    def setOwnerTask(self, task):
        self.task = proxy(task)


    def run(self):
        if not self.task:
            raise se.InvalidRecovery("Recovery - %s: no parent task" % str(self))
        self.validateName(self.object)
        self.validateName(self.function)
        if self.callback:
            self.callback(self)
        # instantiate an object of class "self.object" (bad name)
        module = __import__('storage.' + self.moduleName, locals(), globals(), [self.moduleName])
        classObj = getattr(module, self.object)
        function = getattr(classObj, self.function)
        argslist = self.params.getList()
        return function(self.task, *argslist)


    def __str__(self):
        return "%s: %s->%s(%s)" % (self.name, self.object, self.function, str(self.params))



class TaskResult(object):
    fields = {
        "code": int,
        "message": str,
        "result": str,
    }

    def __init__(self, code=0, message="", result=""):
        self.code = code
        self.message = message
        self.result = result


    def toDict(self):
        return dict(message=self.message, code=str(self.code), result=self.result)


    def __str__(self):
        return "Task result: %s - %s: %s" % (self.code, self.message, self.result)



class TaskResource(object):
    fields = {
        "namespace": str,
        "name": str,
        "lockType": resource.LockType,
    }

    def __init__(self, namespace, name, lockType):
        self.namespace = namespace
        self.name = name
        self.lockType = lockType


    def key(self):
        return "%s%s%s" % (self.namespace, RESOURCE_SEP, self.name)


    def tuple(self):
        return (self.namespace, self.name, self.lockType)


    def __str__(self):
        return "%s/%s - %s" % (self.namespace, self.name, str(self.lockType))



class TaskResourcesList(dict):
    log = logging.getLogger('irs')

    @staticmethod
    def resToDict(reslist):
        d = {}
        for i in reslist:
            d["%s%s%s" % (i[0], RESOURCE_SEP, i[1])] = i
        return d


    def registerList(self, owner, l):
        self.log.debug("TaskResourcesList: registerList: %s" % self.strlist(l))
        for res in l:
            owner.register(*res.tuple())


    def strlist(self, l):
        return str([r.tuple() for r in l])


    def insertResource(self, namespace, name, locktype):
        key = self.toKey(namespace, name)
        if key in self.keys():
            return # TODO: check that locktype is the same
        self[key] = TaskResource(namespace, name, locktype)


    def allGranted(self, owner):
        rdict = self
        owned = self.resToDict(owner.ownedResources())
        self.log.debug("TaskResourcesList.allGranted: check for %s owned %s", str(rdict), owned)
        missing = {}
        for key in rdict:
            if key not in owned:
                missing[key] = rdict[key]
        if not missing:
            return True
        self.log.debug("TaskResourcesList.allGranted: Missing: %s", self.strlist(missing.values()))
        missingresources = {}
        reqs = self.resToDict(owner.requestedResources())
        for key in missing:
            if key not in reqs:
                missingresources[key] = missing[key]
                break
        if not missing:
            return False    # not granted, but nothing is missing
        # register missing resources
        self.log.debug("TaskResourcesList.allGranted: register: %s", self.strlist(missing.values()))
        self.registerList(owner, missingresources.values())
        return False


    @staticmethod
    def toKey(namespace, name):
        return "%s%s%s" % (namespace, RESOURCE_SEP, name)



class TaskPriority(EnumType):
    low = "low"
    medium = "medium"
    high = "high"



class Task:
    # External Task info
    fields = {
        # field_name: type
        "id": str,
        "name": str,
        "tag": str,
        "store": str,
        "recoveryPolicy": TaskRecoveryType,
        "persistPolicy": TaskPersistType,
        "cleanPolicy": TaskCleanType,
        "priority": TaskPriority,
        "state": State,
        "njobs": int,
        "nrecoveries": int,
        "nresources": int,
        "metadataVersion": int
    }

    log = logging.getLogger('irs')

    def __init__(self, id, name="", tag="", recovery=TaskRecoveryType.none, priority=TaskPriority.low):
        """
        id - Unique ID
        name - human readable name
        persist - persistency type: auto-clean/manual-clean/not-persistent
        """

        if not id:
            id = uuid.generate()
        self.metadataVersion = TASK_METADATA_VERSION
        self.validateID(id)
        self.lock = threading.Lock()
        self.callbackLock = threading.Lock()
        self.id = str(id)
        self.name = name
        self.tag = tag
        self.priority = priority
        self.recoveryPolicy = recovery
        self.persistPolicy = TaskPersistType.none
        self.cleanPolicy = TaskCleanType.auto
        self.store = None
        self.defaultException = None

        self.state = State(State.init)
        self.result = TaskResult(0, "Task is initializing", "")

        self.resOwner = resource.Owner(proxy(self), raiseonfailure=True)
        self.resources = TaskResourcesList()
        self.nresources = 0
        self.error = se.TaskAborted("Unknown error encountered")

        self.mng = None
        self._aborting = False
        self._forceAbort = False
        self.ref = 0

        self.recoveries = []
        self.jobs = []
        self.nrecoveries = 0    # just utility count - used by save/load
        self.njobs = 0          # just utility count - used by save/load


    def _debug(self, msg, *args):
        self.log.debug("Task %s: " % str(self) + msg, *args)


    def __del__(self):
        if not self.state.isDone():
            threading.Thread(target=self._done).start()


    def _done(self):
        self.resOwner.releaseAll()
        if self.cleanPolicy == TaskCleanType.auto:
            self.clean()


    def __state_preparing(self, fromState):
        pass


    def __state_blocked(self, fromState):
        pass


    def __state_acquiring(self, fromState):
        if self.resources.allGranted(self.resOwner):
            self._updateState(State.queued)


    def __state_queued(self, fromState):
        try:
            self.mng.queue(self)
        except Exception, e:
            self._setError(e)
            self.stop()


    def __state_running(self, fromState):
        self._runJobs()


    def __state_finished(self, fromState):
        self._done()


    def __state_aborting(self, fromState):
        if self.ref > 1:
            return
        self._debug("_aborting: recover policy %s", self.recoveryPolicy)
        if self.recoveryPolicy == TaskRecoveryType.auto:
            self._updateState(State.racquiring)
        elif self.recoveryPolicy == TaskRecoveryType.none:
            self._updateState(State.failed)
        else:
            self._updateState(State.waitrecover)


    def __state_waitrecover(self, fromState):
        pass


    def __state_racquiring(self, fromState):
        if self.resources.allGranted(self.resOwner):
            self._updateState(State.recovering)


    def __state_recovering(self, fromState):
        self._recover()


    def __state_raborting(self, fromState):
        if self.ref == 1:
            self._updateState(State.failed)
        else:
            self.log.warn("State was change to 'raborting' when ref was not 1.")


    def __state_recovered(self, fromState):
        self._done()


    def __state_failed(self, fromState):
        self._done()


    def __state_cleaning(self, fromState):
        pass


    def _updateState(self, state, force=False):
        fromState = self.state
        requestedState = state
        if self._aborting:
            if self.state.canAbort():
                state = State.aborting
            elif self.state.canAbortRecovery() and state != State.recovered:
                state = State.raborting
        self._aborting = False
        if requestedState == state:
            self._debug("moving from state %s -> state %s", fromState, state)
        else:
            self._debug("moving from state %s -> state %s instead of %s", fromState, state, requestedState)

        self.state.moveto(state, force)
        if self.persistPolicy == TaskPersistType.auto:
            try:
                self.persist()
            except Exception, e:
                misc.logException(e)
                self.log.warning("Task._updateState: failed persisting task %s" % (self.id))

        fn = getattr(self, "_Task__state_%s" % state)
        fn(fromState)


    def _updateResult(self, code, message, result):
        self.result.result = result
        self.result.code = code
        self.result.message = message


    @classmethod
    def validateID(cls, taskID):
        if not taskID or "." in taskID:
            raise se.InvalidParameterException("taskID", taskID)


    @classmethod
    def _loadMetaFile(cls, filename, obj, fields):
        #cls.log.debug("load file %s read obj %s fields %s", filename, str(obj), fields.keys())
        try:
            for line in open(filename, 'r').readlines():
                #cls.log.debug("-- line: %s", line)
                # process current line
                if line.find(KEY_SEPERATOR) < 0:
                    continue
                parts = line.split(KEY_SEPERATOR)
                if len(parts) != 2:
                    cls.log.warning("Task._loadMetaFile: %s - ignoring line '%s'", filename, line)
                    continue

                field = parts[0].strip()
                value = parts[1].strip()
                if field not in fields:
                    cls.log.warning("Task._loadMetaFile: %s - ignoring field %s in line '%s'", filename, field, line)
                    continue

                ftype = fields[field]
                #cls.log.debug("file %s obj %s field %s ftype %s value %s", filename, str(obj), field, ftype, value)
                setattr(obj, field, ftype(value))
        except Exception, e:
            misc.logException(e)
            raise se.TaskMetaDataLoadError(filename)


    @classmethod
    def _dump(cls, obj, fields):
        lines = []
        for field in fields:
            try:
                value = str(getattr(obj, field))
                if KEY_SEPERATOR in field or KEY_SEPERATOR in value:
                    raise ValueError("field and value cannot include %s character" % KEY_SEPERATOR)
                lines.append("%s %s %s" % (field, KEY_SEPERATOR, value))
            except Exception, e:
                misc.logException(e)
                cls.log.warning("Task._dump: object %s skipping field %s" % (str(obj), field))
        return lines


    @classmethod
    def _saveMetaFile(cls, filename, obj, fields):
        try:
            f = file(filename, "w")
            f.writelines([ l + "\n" for l in cls._dump(obj, fields) ])
            f.flush()
            f.close()
        except Exception, e:
            misc.logException(e)
            raise se.TaskMetaDataSaveError(filename)


    def _loadTaskMetaFile(self, taskDir):
        taskFile = os.path.join(taskDir, self.id + TASK_EXT)
        self._loadMetaFile(taskFile, self, Task.fields)


    def _saveTaskMetaFile(self, taskDir):
        taskFile = os.path.join(taskDir, self.id + TASK_EXT)
        self._saveMetaFile(taskFile, self, Task.fields)


    def _loadJobMetaFile(self, taskDir, n):
        taskFile = os.path.join(taskDir, self.id + JOB_EXT + NUM_SEP + str(n))
        self._loadMetaFile(taskFile, self.jobs[n], Job.fields)


    def _saveJobMetaFile(self, taskDir, n):
        taskFile = os.path.join(taskDir, self.id + JOB_EXT + NUM_SEP + str(n))
        self._saveMetaFile(taskFile, self.jobs[n], Job.fields)


    def _loadRecoveryMetaFile(self, taskDir, n):
        taskFile = os.path.join(taskDir, self.id + RECOVER_EXT + NUM_SEP + str(n))
        self._loadMetaFile(taskFile, self.recoveries[n], Recovery.fields)


    def _saveRecoveryMetaFile(self, taskDir, n):
        taskFile = os.path.join(taskDir, self.id + RECOVER_EXT + NUM_SEP + str(n))
        self._saveMetaFile(taskFile, self.recoveries[n], Recovery.fields)


    def _loadTaskResultMetaFile(self, taskDir):
        taskFile = os.path.join(taskDir, self.id + RESULT_EXT)
        self._loadMetaFile(taskFile, self.result, TaskResult.fields)


    def _saveTaskResultMetaFile(self, taskDir):
        taskFile = os.path.join(taskDir, self.id + RESULT_EXT)
        self._saveMetaFile(taskFile, self.result, TaskResult.fields)


    def _loadResourceMetaFile(self, taskDir, key):
        taskFile = os.path.join(taskDir, key + RESOURCE_EXT)
        self._loadMetaFile(taskFile, self.resources[key], TaskResource.fields)
        self._debug("taskFile %s loaded" % taskFile)


    def _saveResourceMetaFile(self, taskDir, key):
        taskFile = os.path.join(taskDir, key + RESOURCE_EXT)
        self._saveMetaFile(taskFile, self.resources[key], TaskResource.fields)


    def _getResourcesKeyList(self, taskDir):
        keys = []
        for path in  glob.glob(os.path.join(taskDir, "*" + RESOURCE_EXT)):
            filename = os.path.basename(path)
            keys.append(filename[:filename.rfind(RESOURCE_EXT)])
        return keys


    def _load(self, storPath, ext=""):
        self.log.debug("%s: load from %s, ext '%s'", str(self), storPath, ext)
        if self.state != State.init:
            raise se.TaskMetaDataLoadError("task %s - can't load self: not in init state" % str(self))
        taskDir = os.path.join(storPath, str(self.id) + str(ext))
        if not ros.exists(taskDir):
            raise se.TaskDirError("load: no such task dir '%s'" % taskDir)
        oldid = self.id
        self._loadTaskMetaFile(taskDir)
        if self.id != oldid:
            raise se.TaskMetaDataLoadError("task %s: loaded file do not match id (%s != %s)" % (str(self), self.id, oldid))
        if self.state == State.finished:
            self._loadTaskResultMetaFile(taskDir)
        for jn in range (self.njobs):
            self.jobs.append(Job("load", None))
            self._loadJobMetaFile(taskDir, jn)
            self.jobs[jn].setOwnerTask(self)
        for rn in range (self.nrecoveries):
            self.recoveries.append(Recovery("load", "load", "load", "load", ""))
            self._loadRecoveryMetaFile(taskDir, rn)
            self.recoveries[rn].setOwnerTask(self)
        for key in self._getResourcesKeyList(taskDir):
            self.resources[key] = TaskResource("none", "none", "none")
            self._loadResourceMetaFile(taskDir, key)


    def _save(self, storPath):
        origTaskDir = os.path.join(storPath, self.id)
        if not ros.exists(origTaskDir):
            raise se.TaskDirError("_save: no such task dir '%s'" % origTaskDir)
        taskDir = os.path.join(storPath, self.id + TEMP_EXT)
        self._debug("_save: orig %s temp %s", origTaskDir, taskDir)
        if ros.exists(taskDir):
            misc.cleanupdir(taskDir)
        ros.makedirs(taskDir)
        try:
            self.njobs = len(self.jobs)
            self.nrecoveries = len(self.recoveries)
            self._saveTaskMetaFile(taskDir)
            if self.state == State.finished:
                self._saveTaskResultMetaFile(taskDir)
            for jn in range (self.njobs):
                self._saveJobMetaFile(taskDir, jn)
            for rn in range (self.nrecoveries):
                self._saveRecoveryMetaFile(taskDir, rn)
            for key in self.resources:
                self._saveResourceMetaFile(taskDir, key)
        except Exception, e:
            misc.logException(e)
            try:
                misc.cleanupdir(taskDir)
            except:
                self.log.warning("can't remove temp taskdir %s" % taskDir)
            raise se.TaskPersistError("%s persist failed: %s" % (str(self), str(e)))
        ros.rename(origTaskDir, origTaskDir + BACKUP_EXT)
        ros.rename(taskDir, origTaskDir)
        #self._debug("cleaning backup '%s'", origTaskDir + BACKUP_EXT)
        misc.cleanupdir(origTaskDir + BACKUP_EXT)


    def _clean(self, storPath):
        taskDir = os.path.join(storPath, self.id)
        misc.cleanupdir(taskDir)


    def _recoverDone(self):
        # protect agains races with stop/abort
        self._debug("Recover Done: state %s", self.state)
        while True:
            try:
                if self.state == State.recovering:
                    self._updateState(State.recovered)
                elif self.state == State.raborting:
                    self._updateState(State.failed)
                return
            except se.TaskStateTransitionError, e:
                misc.logException(e)


    def _recover(self):
        self._debug("_recover")
        if not self.state == State.recovering:
            raise se.TaskStateError("%s: _recover in state %s" % (str(self), self.state))
        try:
            while self.state == State.recovering:
                rec = self.popRecovery()
                self._debug("running recovery %s", str(rec))
                if not rec:
                    break
                self._run(rec.run)
        except Exception, e:
            misc.logException(e)
            self.log.warning("task %s: recovery failed: %s" % (str(self), str(e)))
            # protect agains races with stop/abort
            try:
                if self.state == State.recovering:
                    self._updateState(State.raborting)
            except se.TaskStateTransitionError, e:
                pass
        self._recoverDone()


    def resourceReleased(self, namespace, resource):
        self._debug("resourceReleased: %s.%s", namespace, resource)
        del self.resources[self.resources.toKey(namespace, resource)]
        self.nresources = len(self.resources)


    def resourceAcquired(self, namespace, resource, locktype):
        # Callback from resource.Owner. May be called by another thread.
        self._incref()
        try:
            self.callbackLock.acquire()
            try:
                self._debug("_resourcesAcquired: %s.%s (%s)", namespace, resource, locktype)
                self.resources.insertResource(namespace, resource, locktype)
                self.nresources = len(self.resources)
                if self.state == State.preparing:
                    return
                if self.state == State.acquiring:
                    self._updateState(State.acquiring)
                elif self.state == State.racquiring:
                    self._updateState(State.racquiring)
                elif self.state == State.blocked:
                    self._updateState(State.preparing)
                elif self.state == State.aborting or self.state == State.raborting:
                    self._debug("resource %s.%s acquired while in state %s", namespace, resource, self.state)
                else:
                    raise se.TaskStateError("acquire is not allowed in state %s" % self.state)
            finally:
                self.callbackLock.release()
        finally:
            self._decref()


    def resourceRegistered(self, namespace, resource, locktype):
        self._incref()
        try:
            self.callbackLock.acquire()
            try:
                # Callback from resource.Owner. May be called by another thread.
                self._debug("_resourcesAcquired: %s.%s (%s)", namespace, resource, locktype)
                # Protect against races with stop/abort
                self.resources.insertResource(namespace, resource, locktype)
                self.nresources = len(self.resources)
                if self.state == State.preparing:
                    self._updateState(State.blocked)
            finally:
                self.callbackLock.release()
        finally:
            self._decref()


    def _setError(self, e = se.TaskAborted("Unknown error encountered")):
        misc.logException(e)
        self.error = e


    def _run(self, fn, *args, **kargs):
        code = 100
        message = "Unknown Error"
        try:
            return fn(*args, **kargs)
        except se.StorageException, e:
            code = e.code
            message = e.message
            self._setError(e)
        except Exception, e:
            message = str(e)
            self._setError(e)
        except:
            self._setError()

        self._debug("Task._run: %s %s %s failed - stopping task", str(self), str(args), str(kargs))
        self.stop()
        raise se.TaskAborted(message, code)


    def _runJobs(self):
        result = ""
        code = 100
        message = "Unknown Error"
        i = 0
        j = None
        try:
            if self.aborting():
                raise se.TaskAborted("shutting down")
            if not self.state == State.running:
                raise se.TaskStateError("%s: can't run Jobs in state %s" % (str(self), (self.state)))
            # for now: result is the last job result, jobs are run sequentially
            for j in self.jobs:
                if self.aborting():
                    raise se.TaskAborted("shutting down")
                self.log.debug("Task.run: running job %s: %s" % (i, str(j)))
                result = self._run(j.run)
                if result is None:
                    result = ""
                i += 1
            j = None
            self._updateResult(0, "%s jobs completed successfuly" % i, result)
            self._updateState(State.finished)
            self.log.debug('Task.run: exit - success: result %s' % result)
            return result
        except se.TaskAborted, e:
            self.log.debug("aborting: %s", str(e))
            code = e.abortedcode
            if not self.aborting():
                self.log.error("Aborted exception but not in aborting state")
                raise
        self._updateResult(code, "jobs number %s: %s failed: %s" % (i, str(j), message), "")


    def _doAbort(self, force=False):
        self._debug("Task._doAbort: force %s" % force)
        self.lock.acquire()
        # Am I really the last?
        if self.ref != 0:
            self.lock.release()
            return
        self.ref += 1
        self.lock.release()
        try:
            try:
                if not self.state.canAbort() and (force and not self.state.canAbortRecover()):
                    self.log.warning("Task._doAbort %s: ignoring - at state %s", str(self), str(self.state))
                    return
                self.resOwner.cancelAll()
                if self.state.canAbort():
                    self._updateState(State.aborting)
                else:
                    self._updateState(State.raborting)
            except se.TaskAborted:
                self._updateState(State.failed)
        finally:
            self.lock.acquire()
            self.ref -= 1
            self.lock.release()
            #If somthing horrible went wrong. Just fail the task.
            if not self.state.isDone():
                self.log.warn("Task exited in non terminal state. Setting tasks as failed.")
                self._updateState(State.failed)


    def _doRecover(self):
        self.lock.acquire()
        # Am I really the last?
        if self.ref != 0:
            self.lock.release()
            raise se.TaskHasRefs(str(self))
        self.ref += 1
        self.lock.release()
        try:
            self._updateState(State.racquiring)
        finally:
            self.lock.acquire()
            self.ref -= 1
            self.lock.release()


    def _incref(self, force=False):
        self.lock.acquire()
        try:
            if self.aborting() and (self._forceAbort or not force):
                raise se.TaskAborted(str(self))

            self.ref += 1
            ref = self.ref
            return ref
        finally:
            self.lock.release()


    def _decref(self, force=False):
        self.lock.acquire()
        self.ref -= 1
        ref = self.ref
        self.lock.release()

        self._debug("ref %d aborting %s", ref, self.aborting())
        if ref == 0 and self.aborting():
            self._doAbort(force)
        return ref


    ###############################################################################################
    # Public Interface                                                                                                                       #
    ###############################################################################################

    def setDefaultException(self, exceptionObj):
        # defaultException must have response method
        if exceptionObj and not hasattr(exceptionObj, "response"):
            raise se.InvalidDefaultExceptionException(str(exceptionObj))
        self.defaultException = exceptionObj


    def setTag(self, tag):
        if KEY_SEPERATOR in tag:
            raise ValueError("tag cannot include %s character" % KEY_SEPERATOR)
        self.tag = str(tag)


    def isDone(self):
        return self.state.isDone()


    def addJob(self, job):
        """
        Add async job to the task. Assumes all resources are acquired or registered.
        """
        if not self.mng:
            raise se.UnmanagedTask(str(self))
        if not isinstance(job, Job):
            raise TypeError("Job param %s(%s) must be Job object" % (repr(job), type(job)))
        if self.state != State.preparing:
            raise Exception("Task.addJob: can't add job in non preparing state (%s)" % self.state)
        if not job.name:
            raise ValueError("Task.addJob: name is required")
        name = job.name
        for j in self.jobs:
            if name == j.name:
                raise ValueError("addJob: name '%s' must be unique" % (name))
        job.setOwnerTask(self)
        self.jobs.append(job)
        self.njobs = len(self.jobs)


    def clean(self):
        if not self.store:
            return
        if not self.isDone():
            raise se.TaskStateError("can't clean in state %s" % self.state)
        self._clean(self.store)


    def pushRecovery(self, recovery):
        """
        Add recovery "job" to the task. Recoveries are commited in FILO order.
        Assumes that all required resources are acquired or registered.
        """
        if not isinstance(recovery, Recovery):
            raise TypeError("recovery param %s(%s) must be Recovery object" % (repr(recovery), type(recovery)))
        if not recovery.name:
            raise ValueError("pushRecovery: name is required")
        name = recovery.name
        for r in self.recoveries:
            if name == r.name:
                raise ValueError("pushRecovery: name '%s' must be unique" % (name))
        recovery.setOwnerTask(self)
        self.recoveries.append(recovery)
        self.persist()


    def replaceRecoveries(self, recovery):
        if not isinstance(recovery, Recovery):
            raise TypeError("recovery param %s(%s) must be Recovery object" % (repr(recovery), type(recovery)))
        if not recovery.name:
            raise ValueError("replaceRecoveries: name is required")
        recovery.setOwnerTask(self)
        rec = []
        rec.append(recovery)
        self.recoveries = rec
        self.persist()


    def popRecovery(self):
        if self.recoveries:
            return self.recoveries.pop()


    def clearRecoveries(self):
        self.recoveries = []
        self.persist()


    def setManager(self, manager):
        # If need be, refactor out to "validateManager" method
        if not hasattr(manager, "queue"):
            raise se.InvalidTaskMng(str(manager))
        self.mng = manager


    def setCleanPolicy(self, clean):
        self.cleanPolicy = TaskCleanType(clean)


    def setPersistence(self, store, persistPolicy = TaskPersistType.auto, cleanPolicy = TaskCleanType.auto):
        self.persistPolicy = TaskPersistType(persistPolicy)
        self.store = store
        self.setCleanPolicy(cleanPolicy)
        if self.persistPolicy != TaskPersistType.none and not self.store:
            raise se.TaskPersistError("no store defined")
        taskDir = os.path.join(self.store, self.id)
        try:
            if not os.path.exists(taskDir):
                os.makedirs(taskDir)
        except Exception, e:
            misc.logException(e)
            raise se.TaskPersistError("%s: cannot access/create taskdir %s: %s" % (str(self), taskDir, str(e)))
        if self.persistPolicy == TaskPersistType.auto and self.state != State.init:
            self.persist()


    def setRecoveryPolicy(self, clean):
        self.recoveryPolicy = TaskRecoveryType(clean)


    def rollback(self):
        self.log.debug('(rollback): enter')
        if self.recoveryPolicy == TaskRecoveryType.none:
            self.log.debug("rollback is skipped")
            return
        if not self.isDone():
            raise se.TaskNotFinished("can't rollback in state %s" % self.state)
        self._doRecover()
        self.log.debug('(rollback): exit')


    def persist(self):
        if self.persistPolicy == TaskPersistType.none:
            return
        if not self.store:
            raise se.TaskPersistError("no store defined")
        if self.state == State.init:
            raise se.TaskStateError("can't persist in state %s" % self.state)
        self._save(self.store)


    @classmethod
    def loadTask(cls, store, taskid):
        t = Task(taskid)
        if os.path.exists(os.path.join(store, taskid)):
            ext = ""
        # TBD: is this the correct order (temp < backup) + should temp be considered at all?
        elif os.path.exists(os.path.join(store, taskid + TEMP_EXT)):
            ext = TEMP_EXT
        elif os.path.exists(os.path.join(store, taskid + BACKUP_EXT)):
            ext = BACKUP_EXT
        else:
            raise se.TaskDirError("loadTask: no such task dir '%s/%s'" % (store, taskid))
        t._load(store, ext)
        return t


    def prepare(self, func, *args, **kwargs):
        try:
            self._incref()
        except se.TaskAborted:
            self._doAbort()
            return
        try:
            self._updateState(State.preparing)
            result = None
            try:
                if func:
                    result = self._run(func, *args, **kwargs)
            except se.TaskAborted, e:
                self.log.info("aborting: %s", str(e))
                code = e.abortedcode

            if self.aborting():
                self._debug("Prepare: aborted: %s", str(self.error))
                self._updateResult(code, "Task prepare failed: %s" % (str(self.error)), "")
                raise self.error

            if self.jobs:
                self._debug("Prepare: %s jobs exist, move to acquiring", self.njobs)
                self._updateState(State.acquiring)
                self.log.debug("(task:prepare) returning")
                return dict(uuid=str(self.id))

            self._debug("finished: %s", result)
            self._updateResult(0, "OK", result)
            self._updateState(State.finished)
            return result
        finally:
            self._decref()


    def commit(self, args = None):
        self.log.debug("(task:commit) Commiting task: %s" % self.id)
        vars.task = self
        try:
            self._incref()
        except se.TaskAborted:
            self._doAbort()
            return
        try:
            self._updateState(State.running)
        finally:
            self._decref()


    def aborting(self):
        return self._aborting or self.state == State.aborting or self.state == State.raborting


    def stop(self, force=False):
        self._debug("stopping in state %s (force %s)", self.state, force)
        self._incref(force)
        try:
            self._aborting = True
            self._forceAbort = force
        finally:
            self._decref(force)


    def recover(self, args = None):
        ''' Do not call this function while the task is actually running. this
            method should only be used to recover tasks state after (vdsmd) restart.
        '''
        self._debug('(recover): recovering: state %s', self.state)
        vars.task = self
        try:
            self._incref(force=True)
        except se.TaskAborted:
            self._doAbort(True)
            return
        try:
            if self.isDone():
                self._debug('(recover): task is done: state %s', self.state)
                return
            # if we are not during recover, just abort
            if self.state.canAbort():
                self.stop()
            # if we waited for recovery - keep waiting
            elif self.state == State.waitrecover:
                pass
            # if we started the recovery - restart it
            elif self.state == State.racquiring or self.state == State.recovering:
                self._updateState(State.racquiring, force=True)
            # else we were during failed recovery - abort it
            else:
                self.stop(force=True)
        finally:
            self._decref(force=True)
        self._debug('(recover): recovered: state %s', self.state)


    def getState(self):
        return str(self.state)


    def getInfo(self):
        return dict(id=self.id, verb=self.name)


    def deprecated_getStatus(self):
        oReturn = {}
        oReturn["taskID"] = self.id
        oReturn["taskState"] = self.state.DEPRECATED_STATE[self.state.state]
        oReturn["taskResult"] = self.state.DEPRECATED_RESULT[self.state.state]
        oReturn["code"] = self.result.code
        oReturn["message"] = self.result.message
        return oReturn


    def getStatus(self):
        oReturn = {}
        oReturn["state"] = {'code': self.result.code, 'message': self.result.message}
        oReturn["task"] = {'id': self.id, 'state': str(self.state)}
        oReturn["result"] = self.result.result
        return oReturn


    def getID(self):
        return self.id


    def getTags(self):
        return self.tag


    def __str__(self):
        return str(self.id)


    def dumpTask(self):
        s = "Task: %s" % self._dump(self, Task.fields)
        i = 0
        for r in self.recoveries:
            s += " Recovery%d: %s" % (i, self._dump(r, Recovery.fields))
            i += 1
        i = 0
        for j in self.jobs:
            s += " Job%d: %s" % (i, self._dump(j, Job.fields))
            i += 1
        for key in self.resources:
            s += " Resource-%s: %s" % (key, self._dump(self.resources[key], TaskResource.fields))
        return s


    def getExclusiveLock(self, namespace, resName, timeout=config.getint('irs', 'task_resource_default_timeout')):
        self.resOwner.acquire(namespace, resName, resource.LockType.exclusive, timeout)


    def getSharedLock(self, namespace, resName, timeout=config.getint('irs', 'task_resource_default_timeout')):
        self.resOwner.acquire(namespace, resName, resource.LockType.shared, timeout)



###################################################################################################
# Unit Tests
###################################################################################################

class Test:

    @classmethod
    def recovery0(cls, recovery, p1, p2):
        print "recovery1 %s p1 %s p2 %s" % (str(recovery), p1, p2)


    @classmethod
    def recovery1(cls, recovery, p1, p2):
        print "recovery2 %s p1 %s p2 %s" % (str(recovery), p1, p2)


    @classmethod
    def recovery2(cls, recovery, p1, p2):
        print "recovery3 %s p1 %s p2 %s" % (str(recovery), p1, p2)


    @classmethod
    def recovery3(cls, recovery, p1, p2):
        print "recovery4 %s p1 %s p2 %s" % (str(recovery), p1, p2)


    def job1(self, p1, p2, p3):
        print "job1: %s %s %s" % (p1, p2, p3)
        return dict(k = "one", b="two")


    def job2(self, p1=None, p2="hello"):
        print "job2: %s %s" % (p1, p2)


# Usage example
if __name__ == "__main__":
    import time

    LOG_FILENAME = '/tmp/tasks.log'
    logging.basicConfig(filename=LOG_FILENAME, level=logging.DEBUG, format="%(asctime)s - %(threadName)s - %(message)s")

    def negative(func, *args, **kwargs):
        print "# Negative test: %s %s %s" % (str(func), str(args), str(kwargs))
        try:
            func(*args, **kwargs)
            raise LException("func succeeded - should fail")
        except LException:
            raise
        except Exception, e:
            print "## failed as expected with ''%s'' - OK" % str(e)
            pass

    class LException(Exception):
        def __init__(self, s=None):
            if s:
                logging.error("**** %s", s)
            Exception.__init__(self, s)

    def printargs(a, b):
        print "Args a) %s b) %s" % (a,b)

    # TODO: Change to tempfile
    store = "/tmp/task.unittest"
    if not os.path.exists(store):
        os.makedirs("/tmp/task.unittest")
    print "# Using %s as store" % store
    print "# Create first task"
    t1 = Task("00-00", "unit-test")
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    print "task: %s" % str(t1)
    t1dump = t1.dumpTask()
    print "t1 dump: %s" % t1dump
    print "# prepare task - auto persisted"
    t1.prepare(printargs, "hello world", "nice work")
    #t1.persist()
    t1dump = t1.dumpTask()
    print "t1 dump: %s" % t1dump

    print "# load second task (from saved task)"
    t2 = Task.loadTask(store, "00-00")
    t2dump = t2.dumpTask()
    print "t2 dump: %s" % t2dump
    if t2dump != t1dump:
        raise LException("t2 '%s' != t1 '%s' after load!" % (t2dump, t1dump))

    print "# clean tasks"
    t2.setCleanPolicy(TaskCleanType.none)
    del t2

    t1.clean()
    del t1

    def addrecoveries(t, n, base):
        print "ADD recoveries %s base %s" % (n, base)
        for i in range(n):
            t.pushRecovery(Recovery("recovery num %s" % i, "Test", "recovery%i" % i, ["param1", "param2"]))
        print "Now FAIL"
        raise Exception("this is my failure message")

    print "# Try adding recoveries"
    t1 = Task("00-01", "unit-test-recovery")
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    negative(t1.prepare, addrecoveries, t1, 2, "kuku")

    if t1.state != State.failed:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.failed))

    print "dump: %s" % t1.dumpTask()
    del t1

    print "# Now load the failed task"
    t2 = Task.loadTask(store, "00-01")
    print t2.dumpTask()
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.clean()

    print "# Try adding recoveries - with auto recovery"
    t1 = Task("00-01", "unit-test-recovery", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    negative(t1.prepare, addrecoveries, t1, 4, "kuku")

    if t1.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.recovered))

    print "dump: %s" % t1.dumpTask()
    del t1

    print "# Try adding recoveries - with manual recovery"
    t1 = Task("00-01", "unit-test-recovery", recovery=TaskRecoveryType.manual)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    negative(t1.prepare, addrecoveries, t1, 4, "kuku")

    if t1.state != State.waitrecover:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.waitrecover))

    print '# Do the rollback(recovery)'
    t1.rollback()

    if t1.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.recovered))

    print "dump: %s" % t1.dumpTask()
    del t1

    print "# Try adding recoveries - with manual recovery, rollback after load"
    t1 = Task("00-01", "unit-test-recovery", recovery=TaskRecoveryType.manual)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    negative(t1.prepare, addrecoveries, t1, 4, "kuku")

    if t1.state != State.waitrecover:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.waitrecover))

    del t1

    print '# Load from store'
    t2 = Task.loadTask(store, "00-01")
    print t2.dumpTask()
    t2.setPersistence(store, cleanPolicy=TaskCleanType.auto)
    print '# Do the rollback(recovery)'
    t2.rollback()
    t2.clean()

    if t2.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t2), t2.state, State.recovered))

    print "dump: %s" % t2.dumpTask()
    del t2

    print "# Test adding recoveries - with auto recovery, recovery will fail"
    t1 = Task("00-01", "unit-test-recovery", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    # too many recoveries registered...
    negative(t1.prepare, addrecoveries, t1, 8, "kuku")

    if t1.state != State.failed:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.failed))

    print "dump: %s" % t1.dumpTask()
    del t1

    test = Test()

    def addjob(t, param):
        print "ADD jobs param %s" % (param)
        t.addJob(Job("this is the first job", test.job1, "nice", param, p3="not me"))

    class Manager(object):
        def queue(self, t):
            print "Manager: request to queue %s" % t

    mng = Manager()

    print "# Test adding jobs"
    t1 = Task("00-01", "unit-test-recovery", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(addjob, t1, "kukujob")

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    print "dump: %s" % t1.dumpTask()

    print "now commit the task (run jobs)"
    t1.commit()

    if t1.state != State.finished:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.finished))

    print "dump: %s" % t1.dumpTask()
    del t1

    def addseveraljobs(t, param):
        print "ADD several jobs param %s" % (param)
        t.addJob(Job("this is the first job", test.job1, "nice", param, p3="not me"))
        t.addJob(Job("this is the second job", test.job1, "again", param, p3="just me"))
        t.addJob(Job("this is the third job", test.job2, "nicejob", p2="not me"))

    print "# Test adding several jobs"
    t1 = Task("00-02", "unit-test-recovery-several-jobs", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(addseveraljobs, t1, "kukujob-yeh")

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    print "dump: %s" % t1.dumpTask()

    print "now commit the task (run jobs)"
    t1.commit()

    if t1.state != State.finished:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.finished))

    print "dump: %s" % t1.dumpTask()
    del t1

    def addjobsandrecs(t, n, param):
        print "ADD several jobs %s recoveries, param %s" % (n, param)
        for i in range(n):
            t.pushRecovery(Recovery("recovery num %s" % i, "Test", "recovery%i" % i, ["param1", "param2"]))
        t.addJob(Job("this is the first job", test.job1, "nice", param, p3="not me"))
        t.addJob(Job("this is the second job", test.job1, "again", param, p3="just me"))
        t.addJob(Job("this is the third job", test.job2, "nicejob", p2="not me"))

    print "# Test adding several jobs - with recoveries"
    t1 = Task("00-03", "unit-test-recovery-several-jobs and recoveries", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(addjobsandrecs, t1, 3, "yeh-yeh-yeh")

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    print "dump: %s" % t1.dumpTask()

    print "now commit the task (run jobs)"
    t1.commit()

    if t1.state != State.finished:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.finished))

    print "dump: %s" % t1.dumpTask()
    del t1

    print "Try to load the last one - and roll it back"
    t2 = Task.loadTask(store, "00-03")
    print t2.dumpTask()
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    print '# Do the rollback(recovery)'
    t2.rollback()
    t2.clean()

    if t2.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t2), t2.state, State.recovered))

    print "dump: %s" % t2.dumpTask()
    del t2

    def longwait(sec):
        logging.debug("long wait: sec %s starting now", sec)
        time.sleep(sec)
        logging.debug("long wait: sec %s ENDED", sec)

    def dofail(fail=True):
        logging.debug("dofail: %s", fail)
        if fail:
            raise Exception("Must fail...")

    def complex(t, n, fail, wait):
        print "ADD several register for several resources, jobs %s recoveries, fail %s wait %s" % (n, fail, wait)
        t.resOwner.acquire("testns", "res1", resource.LockType.shared, 6000)
        print "resource testns/res1 - acquired"
        t.resOwner.acquire("testns", "res2", resource.LockType.exclusive, 6000)
        print "resource testns/res2 - acquired"
        t.resOwner.register("testns", "resI", resource.LockType.shared)
        print "resource testns/resI - register"
        t.resOwner.register("testns", "resII", resource.LockType.exclusive)
        print "resource testns/resII acquired/registed"
        for i in range(n):
            t.pushRecovery(Recovery("recovery num %s" % i, "Test", "recovery%i" % i, ["param1", "param2"]))
        t.addJob(Job("this is the first job", test.job1, "nice", n, p3="not me"))
        t.addJob(Job("this is the second job", test.job1, "again", n, p3="just me"))
        t.addJob(Job("this is the third job", test.job2, "nicejob", p2="not me"))
        t.addJob(Job("long task", longwait, wait))
        t.addJob(Job("fail task", dofail, fail))

    print "# Test adding several jobs - with recoveries and resources"
    resource.Namespace("testns")

    t2 = Task("00-04", "unit-test-own-resources", recovery=TaskRecoveryType.auto)
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.setManager(mng)
    t2.prepare(complex, t2, 3, fail=False, wait=2)

    if t2.state != State.queued:
        raise LException("%s task state is '%s' instead of '%s'" % (str(t2), t2.state, State.queued))
    threading.Timer(2, t2.commit).start()

    t1 = Task("00-05", "unit-test-complex-block-on res", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(complex, t1, 3, fail=False, wait=3)

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    print "dump: %s" % t1.dumpTask()

    print "now commit the task (run jobs)"
    t1.commit()

    if t1.state != State.finished:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.finished))

    print "dump: %s" % t1.dumpTask()
    del t1

    print "# Test adding several jobs - with recoveries and resources + recover after lock failure (timeout)"

    t2 = Task("00-06", "unit-test-own-resources", recovery=TaskRecoveryType.auto)
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.setManager(mng)
    t2.prepare(complex, t2, 3, fail=False, wait=8)

    if t2.state != State.queued:
        raise LException("%s task state is '%s' instead of '%s'" % (str(t2), t2.state, State.queued))
    threading.Timer(2, t2.commit).start()

    t1 = Task("00-07", "unit-test-complex-block-on res", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    negative(t1.prepare, complex, t1, 3, False, 3)

    # timed out after 6 sec, and resource is released after 8 + 2 (for commit), wait another 5
    time.sleep(5)
    if t1.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.recovered))

    print "dump: %s" % t1.dumpTask()

    del t1

    print "# Test adding several jobs - with recoveries and resources + recovery + load"

    t2 = Task("00-08", "unit-test-own-resources", recovery=TaskRecoveryType.auto)
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.setManager(mng)
    t2.prepare(complex, t2, 3, fail=False, wait=2)

    if t2.state != State.queued:
        raise LException("%s task state is '%s' instead of '%s'" % (str(t2), t2.state, State.queued))
    threading.Timer(2, t2.commit).start()

    t1 = Task("00-09", "unit-test-complex-block-on res", recovery=TaskRecoveryType.manual)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(complex, t1, 3, fail=True, wait=3)

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    print "dump: %s" % t1.dumpTask()

    print "now commit the task (run jobs)"
    t1.commit()

    if t1.state != State.waitrecover:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.waitrecover))

    print "dump: %s" % t1.dumpTask()
    time.sleep(6)
    t1._done()
    del t1
    del t2

    print "Try to load the last one - and roll it back"
    t2 = Task.loadTask(store, "00-09")
    print t2.dumpTask()
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.setRecoveryPolicy(TaskRecoveryType.auto)
    print '# Do the rollback(recovery)'
    t2.recover()
    t2.rollback()

    if t2.state != State.recovered:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t2), t2.state, State.recovered))

    print "dump: %s" % t2.dumpTask()
    del t2

    print "# Test adding several jobs - with recoveries and resources + stop task"

    t2 = Task("00-0a", "unit-test-stop", recovery=TaskRecoveryType.auto)
    t2.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t2.setManager(mng)
    t2.prepare(complex, t2, 3, fail=False, wait=8)

    if t2.state != State.queued:
        raise LException("%s task state is '%s' instead of '%s'" % (str(t2), t2.state, State.queued))
    threading.Timer(2, t2.stop).start()
    threading.Timer(3, t2._runJobs).start()

    t1 = Task("00-0b", "unit-test-after-stop", recovery=TaskRecoveryType.auto)
    t1.setPersistence(store, cleanPolicy=TaskCleanType.manual)
    t1.setManager(mng)
    t1.prepare(complex, t1, 3, False, 3)

    if t1.state != State.queued:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.queued))

    t1.commit()
    print "dump: %s" % t1.dumpTask()

    if t1.state != State.finished:
        raise LException("%s task failed without recovery - but state is '%s' instead of '%s'" % (str(t1), t1.state, State.finished))

    del t1

    print "Done"
